package com.k8.nacosRegistry;

import com.alibaba.nacos.api.annotation.NacosProperties;
import com.alibaba.nacos.api.config.*;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.client.config.listener.impl.AbstractConfigChangeListener;
import com.alibaba.nacos.client.env.NacosClientProperties;
import com.k8.common.factroy.BeanFactory;
import com.k8.common.util.StrUtil;
import com.k8.registry.RegistryException;
import com.k8.registry.listener.InstanceListener;
import com.k8.registry.impl.ListenerRegistry;
import com.k8.rpc.metadata.AppMetaData;
import com.k8.rpc.metadata.ProtocolMetadata;
import com.k8.rpc.url.ApiSignature;
import com.k8.rpc.url.RegistrySignature;
import com.k8.rpc.url.URL;
import com.k8.rpc.util.RpcUtils;
import com.k8.rpc.util.URLUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import static com.alibaba.nacos.api.annotation.NacosProperties.PASSWORD;
import static com.alibaba.nacos.api.annotation.NacosProperties.USERNAME;
import static com.k8.common.constants.CommonConstants.*;
import static com.k8.common.constants.CommonKeyConstants.CLUSTER_NAME_KEY;
import static com.k8.common.constants.CommonKeyConstants.CLUSTER_SET_KEY;
import static com.k8.common.constants.CommonKeyConstants.RegistryKeyConstant.*;
import static com.k8.common.constants.CommonKeyConstants.RpcKeyConstant.EXPORTER_SIGNATURE_KEY;

public class NacosRegistry extends ListenerRegistry {
    private static final Logger logger = LoggerFactory.getLogger(NacosRegistry.class);
    private static final String DEFAULT_USERNAME = "nacos";
    private static final String DEFAULT_PASSWORD = "nacos";
    private static final String IP = "ip";
    private static final String PORT = "port";
    private static final String NACOS_SERVER_ADDR_KEY = "serverAddr";

    // 缓存NamingService和ConfigService，key为服务地址
    private final Map<String, NamingService> namingServiceCache = new ConcurrentHashMap<>();
    private final Map<String, ConfigService> configServiceCache = new ConcurrentHashMap<>();

    // 缓存订阅关系，用于取消订阅
    private final Map<String, Set<EventListener>> subscriptionCache = new ConcurrentHashMap<>();
    private final ReentrantLock lock = new ReentrantLock();

    @Override
    public void unregister(URL url) {
        super.unregister(url);
        doUnregisterInstance(url);
    }

    /**
     * 注册需要实现对路由的注册
     * prototype 可以通过 {@link com.alibaba.nacos.api.PropertyKeyConst}中的key进行配置f
     */
    @Override
    protected void doRegisterInstance(Object appMetaData, URL registerUrl) {
        // 创建新的NacosClientProperties实例，避免线程安全问题
        NacosClientProperties properties = createNacosClientProperties(registerUrl);
        Instance instance = new Instance();
        initInstance(instance, registerUrl, appMetaData);
        try {
            NamingService namingService = getNamingService(registerUrl.getAddress().getAddr(), properties);
            String group = registerUrl.getParameter(INSTANCE_GROUP_KEY, K8_GROUP);
            namingService.registerInstance(registerUrl.getAppName(), group, instance);
            logger.info("Successfully registered instance: {}:{} for service: {}",
                    instance.getIp(), instance.getPort(), registerUrl.getAppName());
        } catch (NacosException e) {
            logger.error("Failed to register instance to Nacos", e);
            throw new RegistryException("Failed to register instance to Nacos");
        }
    }

    /**
     * 服务注销实现
     */
    protected void doUnregisterInstance(URL url) {
        NacosClientProperties properties = createNacosClientProperties(url);

        try {
            NamingService namingService = getNamingService(url.getAddress().getAddr(), properties);
            String group = url.getParameter(INSTANCE_GROUP_KEY, K8_GROUP);

            // 创建实例对象用于注销
            Instance instance = new Instance();
            initInstance(instance, url, null);

            namingService.deregisterInstance(url.getAppName(), group, instance);
            logger.info("Successfully unregistered instance: {}:{} for service: {}",
                    instance.getIp(), instance.getPort(), url.getAppName());
        } catch (NacosException e) {
            logger.error("Failed to unregister instance from Nacos", e);
            throw new RegistryException("Failed to unregister instance from Nacos");
        }
    }

    /**
     * 获取远程服务实例并订阅服务变化
     */
    @Override
    protected List<AppMetaData> doGetRemoteAppInstancesAndSubscribe(
            String mapping, ApiSignature apiSignature, URL registryConfigUrl,
            URL baseProtocolUrl, RegistryEventDeployerWrapper registryEventDeployerWrapper) {

        try {
            String serviceAddress = registryConfigUrl.getAddress().getAddr();
            NamingService namingService = getNamingService(serviceAddress, createNacosClientProperties(registryConfigUrl));

            String group = registryConfigUrl.getParameter(INSTANCE_GROUP_KEY, K8_GROUP);
            String clusters = registryConfigUrl.getParameter(CLUSTER_SET_KEY, "DEFAULT");

            List<String> clusterList = parseClusters(clusters);

            // 添加集群订阅
            List<Instance> allInstances = namingService.getAllInstances(mapping, group, clusterList, true);

            // 创建监听器并缓存
            NacosInstanceListener listener = new NacosInstanceListener(
                    this, baseProtocolUrl, registryEventDeployerWrapper, apiSignature.getSignature());
            String cacheKey = getSubscriptionCacheKey(mapping, group, clusterList);

            lock.lock();
            try {
                subscriptionCache.computeIfAbsent(cacheKey, k -> ConcurrentHashMap.newKeySet()).add(listener);
            } finally {
                lock.unlock();
            }

            namingService.subscribe(mapping, group, clusterList, listener);

            return allInstances.stream()
                    .map(instance -> restoreMetadata(instance.getMetadata()))
                    .toList();
        } catch (NacosException e) {
            logger.error("Failed to get remote instances and subscribe", e);
            throw new RegistryException("Failed to get remote instances and subscribe");
        }
    }

    /**
     * 取消订阅
     */
    public void unsubscribe(String mapping, String group, List<String> clusters) {
        String cacheKey = getSubscriptionCacheKey(mapping, group, clusters);
        Set<EventListener> listeners = subscriptionCache.remove(cacheKey);

        if (listeners != null && !listeners.isEmpty()) {
            try {
                // 假设使用默认配置的注册中心地址
                URL registryUrl = getUrl();
                NamingService namingService = getNamingService(registryUrl.getAddress().getAddr(),
                        createNacosClientProperties(registryUrl));

                for (EventListener listener : listeners) {
                    namingService.unsubscribe(mapping, group, clusters, listener);
                }
                logger.info("Unsubscribed from service: {} with group: {} and clusters: {}",
                        mapping, group, clusters);
            } catch (NacosException e) {
                logger.error("Failed to unsubscribe from service", e);
            }
        }
    }

    /**
     * 创建Nacos客户端属性，每次创建新实例避免线程安全问题
     */
    private NacosClientProperties createNacosClientProperties(URL url) {
        NacosClientProperties properties = NacosClientProperties.PROTOTYPE.derive();

        // 设置URL中的参数
        url.getUrlParams().forEach((key, value) -> {
            if (value != null) {
                properties.setProperty(key, value);
            }
        });

        // 设置服务器地址
        properties.setProperty(NACOS_SERVER_ADDR_KEY, url.getAddress().getAddr());

        // 设置默认用户名密码（如果URL中没有指定）
        if (!properties.containsKey(USERNAME)) {
            properties.setProperty(USERNAME, DEFAULT_USERNAME);
        }
        if (!properties.containsKey(PASSWORD)) {
            properties.setProperty(PASSWORD, DEFAULT_PASSWORD);
        }

        return properties;
    }

    /**
     * 初始化服务实例信息
     */
    private void initInstance(Instance instance, URL url, Object appMetaData) {
        int port = url.getPort(8080);
        String ip = url.getIp();
        String cluster = url.getParameter(CLUSTER_NAME_KEY, "DEFAULT");

        instance.setPort(port);
        // 使用IP:PORT:VERSION作为唯一实例ID，避免冲突
        instance.setInstanceId(ip + ":" + port + ":" + url.getVersion());
        instance.setClusterName(cluster);
        instance.setIp(ip);
        instance.setServiceName(url.getAppName());
        instance.setHealthy(true);
        instance.setWeight(url.getParameter(INSTANCE_WEIGHT_KEY, 1));
        if (appMetaData != null) {
            instance.setMetadata((Map<String, String>) appMetaData);
        }
    }

    /**
     * 注册映射和配置信息
     */
    @Override
    public void doRegisterMappingAndConfiguration(URL url, String configure, ApiSignature apiSignature) {
        String appName = url.getAppName();
        String interfaceName = apiSignature.getInterfaceName();
        String configureDataId = apiSignature.getSignature();

        NacosClientProperties properties = createNacosClientProperties(url);
        String configGroup = url.getParameter(CONFIGURE_GROUP_KEY, K8_GROUP);

        try {
            ConfigService configService = getConfigService(url.getAddress().getAddr(), properties);

            // 注册映射信息
            configService.publishConfig(interfaceName,
                    url.getParameter(MAPPING_GROUP_KEY, MAPPING_GROUP),
                    appName);

            // 校验配置格式并发布
            if (ConfigType.PROPERTIES.getType().equals(url.getParameter(CONFIG_TYPE_KEY, ConfigType.PROPERTIES.getType()))) {
                if (!isValidPropertiesFormat(configure)) {
                    throw new IllegalArgumentException("Configure content is not valid properties format");
                }
            }

            configService.publishConfig(configureDataId, configGroup, configure,
                    url.getParameter(CONFIG_TYPE_KEY, ConfigType.PROPERTIES.getType()));

            logger.info("Successfully registered mapping and configuration for service: {}", interfaceName);
        } catch (NacosException e) {
            logger.error("Failed to register mapping and configuration", e);
            throw new RegistryException("Failed to register mapping and configuration");
        }
    }

    /**
     * 获取服务提供者的远程映射
     */
    @Override
    protected String getRemoteMappingOfServiceToProvider(ApiSignature apiSignature, URL url) {
        String interfaceName = apiSignature.getInterfaceName();

        try {
            ConfigService configService = getConfigService(url.getAddress().getAddr(), createNacosClientProperties(url));
            return configService.getConfig(interfaceName,
                    url.getParameter(MAPPING_GROUP_KEY, MAPPING_GROUP), 5000);
        } catch (NacosException e) {
            logger.error("Failed to get remote mapping for service: {}", interfaceName, e);
            throw new RegistryException("Failed to get remote mapping");
        }
    }

    /**
     * 获取远程协议服务配置
     */
    @Override
    protected String getRemoteProtocolServiceConfigure(ApiSignature apiSignature, URL url) {
        String configureDataId = apiSignature.getSignature();
        String configGroup = url.getParameter(CONFIGURE_GROUP_KEY, K8_GROUP);

        try {
            ConfigService configService = getConfigService(url.getAddress().getAddr(), createNacosClientProperties(url));
            return configService.getConfig(configureDataId, configGroup, 5000);
        } catch (NacosException e) {
            logger.error("Failed to get remote protocol configuration for: {}", configureDataId, e);
            throw new RegistryException("Failed to get remote protocol configuration");
        }
    }

    /**
     * 将配置还原到URL中
     */
    protected void restoreConfigureToURL(String configure, URL protocolURL) {
        if (StrUtil.isEmpty(configure)) {
            logger.warn("Empty configure to restore to URL");
            return;
        }

        Properties properties = new Properties();
        try (ByteArrayInputStream input = new ByteArrayInputStream(configure.getBytes())) {
            properties.load(input);
            for (String key : properties.stringPropertyNames()) {
                protocolURL.setParameter(key, properties.getProperty(key));
            }
        } catch (IOException e) {
            logger.error("Failed to restore configure to URL", e);
            throw new RegistryException("Failed to restore configure to URL");
        }
    }

    /**
     * 将URL参数转换为配置字符串
     */
    protected String doGetConfigure(URL url) {
        Properties properties = new Properties();
        Map<String, String> urlParams = url.getUrlParams();

        urlParams.forEach(properties::put);

        try (ByteArrayOutputStream writer = new ByteArrayOutputStream()) {
            properties.store(writer, null);
            return writer.toString("UTF-8");
        } catch (IOException e) {
            logger.error("Failed to get configure from URL", e);
            throw new RegistryException("Failed to get configure from URL");
        }
    }

    /**
     * 解析集群配置
     */
    private List<String> parseClusters(String clusters) {
        if (StrUtil.isEmpty(clusters) || "DEFAULT".equals(clusters)) {
            return Collections.singletonList("DEFAULT");
        }

        return Arrays.stream(clusters.split(CLUSTER_SET_SPLIT))
                .map(String::trim)
                .filter(StrUtil::isNotEmpty)
                .toList();
    }

    /**
     * 验证Properties格式
     */
    private boolean isValidPropertiesFormat(String content) {
        if (StrUtil.isEmpty(content)) {
            return true; // 空内容视为有效
        }

        // 按行分割（兼容Windows和Linux换行符）
        String[] lines = content.split("\\r?\\n");

        // 用于判断是否存在有效键值对（避免全是注释/空行的情况）
        boolean hasValidKvp = false;

        for (String line : lines) {
            String trimmedLine = line.trim();

            // 空行：跳过
            if (trimmedLine.isEmpty()) {
                continue;
            }

            // 注释行（以#或!开头）：跳过
            if (trimmedLine.startsWith("#") || trimmedLine.startsWith("!")) {
                continue;
            }

            // 验证键值对格式
            int equalsIndex = trimmedLine.indexOf('=');
            // 不包含=号，或=号在开头（无键）
            if (equalsIndex <= 0) {
                logger.warn("Invalid properties line: {}", line);
                return false;
            }

            // 提取键（等号前的部分）
            String key = trimmedLine.substring(0, equalsIndex).trim();
            // 键不能为空，且只能包含字母、数字、下划线、横线、点、冒号
            if (key.isEmpty() || !key.matches("^[a-zA-Z0-9_\\-.:]+$")) {
                logger.warn("Invalid key in properties line: {}", line);
                return false;
            }

            // 提取值（等号后的部分，允许为空）
            String value = trimmedLine.substring(equalsIndex + 1).trim();
            // 值可以为空，但不能包含非法字符（可选，根据业务需求调整）
            // if (value.contains("某些不允许的字符")) { ... }

            hasValidKvp = true;
        }

        // 至少需要有一个有效的键值对
        return hasValidKvp;
    }

    /**
     * 获取订阅缓存的key
     */
    private String getSubscriptionCacheKey(String serviceName, String group, List<String> clusters) {
        return serviceName + ":" + group + ":" + String.join(",", clusters);
    }

    /**
     * 获取或创建NamingService（带缓存）
     */
    private NamingService getNamingService(String serviceAddress, NacosClientProperties properties) throws NacosException {
        return namingServiceCache.computeIfAbsent(serviceAddress, addr -> {
            try {
                return NamingFactory.createNamingService(properties.asProperties());
            } catch (NacosException e) {
                logger.error("Failed to create NamingService for address: {}", addr, e);
                throw new RegistryException("Failed to create NamingService");
            }
        });
    }

    /**
     * 获取或创建ConfigService（带缓存）
     */
    private ConfigService getConfigService(String serviceAddress, NacosClientProperties properties) throws NacosException {
        return configServiceCache.computeIfAbsent(serviceAddress, addr -> {
            try {
                return ConfigFactory.createConfigService(properties.asProperties());
            } catch (NacosException e) {
                logger.error("Failed to create ConfigService for address: {}", addr, e);
                throw new RegistryException("Failed to create ConfigService");
            }
        });
    }

    /**
     * 关闭所有服务，释放资源
     * todo
     */
    public void close() {
        // 关闭NamingService
        namingServiceCache.values().forEach(service -> {
            try {
                service.shutDown();
            } catch (NacosException e) {
                logger.warn("Error while shutting down NamingService", e);
            }
        });
        namingServiceCache.clear();

        // 关闭ConfigService
        configServiceCache.values().forEach(service -> {
            try {
                service.shutDown();
            } catch (Exception e) {
                logger.warn("Error while shutting down ConfigService", e);
            }
        });
        configServiceCache.clear();

        // 清除订阅缓存
        subscriptionCache.clear();
    }

    public NacosRegistry(URL url, BeanFactory beanFactory) {
        super(url, beanFactory);
    }

    @Override
    protected Map<String, String> resolveMetadata(AppMetaData metaData) {
        return convertAppMetaDataToMap(metaData);
    }

    @Override
    protected AppMetaData restoreMetadata(Object metadataObj) {
        if (!(metadataObj instanceof Map)) {
            logger.warn("Invalid metadata type, expected Map but got: {}", metadataObj.getClass().getName());
            return new AppMetaData();
        }

        @SuppressWarnings("unchecked")
        Map<String, String> metaData = (Map<String, String>) metadataObj;
        return parseMapToAppMetaData(metaData);
    }

    /**
     * 将AppMetaData转换为Map
     */
    private static Map<String, String> convertAppMetaDataToMap(AppMetaData appMetaData) {
        Map<String, String> metadataMap = new HashMap<>();

        // 基础字段
        if (appMetaData.getServiceInstanceId() != null) {
            metadataMap.put("serviceInstanceId", String.valueOf(appMetaData.getServiceInstanceId()));
        }
        metadataMap.put("ip", appMetaData.getIp());
        if (appMetaData.getPort() != null) {
            metadataMap.put("port", String.valueOf(appMetaData.getPort()));
        }

        // 处理嵌套的protocols Map
        if (appMetaData.getProtocols() != null) {
            for (Map.Entry<String, ProtocolMetadata> entry : appMetaData.getProtocols().entrySet()) {
                String protocolKeyPrefix = "protocols." + entry.getKey() + ".";
                Map<String, String> protocolMap = convertProtocolMetadataToMap(entry.getValue());
                protocolMap.forEach((k, v) -> metadataMap.put(protocolKeyPrefix + k, v));
            }
        }

        return metadataMap;
    }

    /**
     * 将ProtocolMetadata转换为Map
     */
    private static Map<String, String> convertProtocolMetadataToMap(ProtocolMetadata protocol) {
        Map<String, String> map = new HashMap<>();
        map.put("protocol", protocol.getProtocol());
        if (protocol.getPort() != null) {
            map.put("port", String.valueOf(protocol.getPort()));
        }
        return map;
    }

    /**
     * 将Map解析为AppMetaData
     */
    public static AppMetaData parseMapToAppMetaData(Map<String, String> metadataMap) {
        AppMetaData appMetaData = new AppMetaData();

        // 解析基础字段
        try {
            if (metadataMap.containsKey("serviceInstanceId")) {
                appMetaData.setServiceInstanceId(Long.parseLong(metadataMap.get("serviceInstanceId")));
            }
        } catch (NumberFormatException e) {
            logger.warn("Invalid serviceInstanceId format: {}", metadataMap.get("serviceInstanceId"), e);
        }

        appMetaData.setIp(metadataMap.get("ip"));

        try {
            if (metadataMap.containsKey("port")) {
                appMetaData.setPort(Integer.parseInt(metadataMap.get("port")));
            }
        } catch (NumberFormatException e) {
            logger.warn("Invalid port format: {}", metadataMap.get("port"), e);
        }

        // 解析嵌套的protocols Map
        Map<String, ProtocolMetadata> protocols = new HashMap<>();
        for (String key : metadataMap.keySet()) {
            if (key.startsWith("protocols.")) {
                String[] parts = key.split("\\.");
                if (parts.length == 3) {
                    String protocolName = parts[1];
                    String fieldName = parts[2];

                    ProtocolMetadata protocol = protocols.computeIfAbsent(protocolName, k -> new ProtocolMetadata());
                    String value = metadataMap.get(key);

                    if ("protocol".equals(fieldName)) {
                        protocol.setProtocol(value);
                    } else if ("port".equals(fieldName)) {
                        try {
                            protocol.setPort(Integer.parseInt(value));
                        } catch (NumberFormatException e) {
                            logger.warn("Invalid protocol port format: {} for protocol: {}", value, protocolName, e);
                        }
                    }
                }
            }
        }
        appMetaData.setProtocols(protocols);

        return appMetaData;
    }

    /**
     * Nacos实例监听器，用于转发事件
     */
    public static class NacosInstanceListener implements EventListener {
        private static final Logger listenerLogger = LoggerFactory.getLogger(NacosInstanceListener.class);
        private final NacosRegistry registry;
        private final URL baseProtocolUrl;
        private final String apiSignature;
        private final RegistryEventDeployerWrapper registryEventDeployerWrapper;

        public NacosInstanceListener(NacosRegistry registry, URL baseProtocolUrl,
                                     RegistryEventDeployerWrapper registryEventDeployerWrapper, String apiSignature) {
            this.registry = registry;
            this.registryEventDeployerWrapper = registryEventDeployerWrapper;
            this.baseProtocolUrl = baseProtocolUrl;
            this.apiSignature = apiSignature;
        }

        @Override
        public void onEvent(Event event) {
            if (event instanceof NamingEvent namingEvent) {
                try {
                    List<Instance> instances = namingEvent.getInstances();
                    Map<Long, AppMetaData> idMaps = new HashMap<>(instances.size());

                    for (Instance instance : instances) {
                        AppMetaData appMetaData = registry.restoreMetadata(instance.getMetadata());
                        Long serviceInstanceId = appMetaData.getServiceInstanceId();
                        if (serviceInstanceId != null) {
                            idMaps.put(serviceInstanceId, appMetaData);
                        } else {
                            listenerLogger.warn("Instance {}:{} has no serviceInstanceId",
                                    instance.getIp(), instance.getPort());
                        }
                    }

                    URL url = new URL(baseProtocolUrl);
                    InstanceListener.InstanceEvent instanceEvent =
                            new InstanceListener.InstanceEvent(url, idMaps, apiSignature);
                    registryEventDeployerWrapper.publish(instanceEvent);
                } catch (Exception e) {
                    listenerLogger.error("Error processing NamingEvent", e);
                }
            }
        }
    }

    /**
     * Nacos配置变更监听器
     */
    @Deprecated
    public static class NacosConfigChangeListener extends AbstractConfigChangeListener {
        private final Logger configLogger = LoggerFactory.getLogger(NacosConfigChangeListener.class);
        private final ListenerRegistry.RegistryEventDeployerWrapper registryEventDeployerWrapper;
        private final URL registryConfigUrl;
        private final String exporterSignature;

        public NacosConfigChangeListener(ListenerRegistry.RegistryEventDeployerWrapper registryEventDeployerWrapper,
                                         URL registryConfigUrl, ApiSignature apiSignature) {
            this.registryEventDeployerWrapper = registryEventDeployerWrapper;
            this.registryConfigUrl = registryConfigUrl;
            this.exporterSignature = RpcUtils.getExporterSignature(apiSignature,
                    new RegistrySignature(registryConfigUrl));
        }

        @Override
        public void receiveConfigChange(ConfigChangeEvent event) {
            try {
                configLogger.info("Registry config changed, registry: {}",
                        registryConfigUrl.getAddress().getAddr());

                Collection<ConfigChangeItem> changeItems = event.getChangeItems();
                URL url = URLUtils.copySimpleUrl(registryConfigUrl);
                url.putAttribute(EXPORTER_SIGNATURE_KEY, exporterSignature);

                for (ConfigChangeItem changeItem : changeItems) {
                    url.setParameter(changeItem.getKey(), changeItem.getNewValue());
                }

                // registryEventDeployerWrapper.publishUrlToExportAsync(url);
            } catch (Exception e) {
                configLogger.error("Error processing config change event", e);
            }
        }
    }
}
